package com.diao.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * @author: chenzhidiao
 * @date: 2021/1/11 8:40
 * @description:
 * @version: 1.0
 */
public class WordCountDemo {
    private static final Logger logger = LoggerFactory.getLogger(WordCountDemo.class);
    public static void main(String[] args) throws Exception {

        //步骤一：初始化程序入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //步骤二：数据的输入
        String hostname = ParameterTool.fromArgs(args).get("hostname");
        int port = ParameterTool.fromArgs(args).getInt("port");
        DataStreamSource<String> dataStream = env.socketTextStream(hostname, port).setParallelism(1);
        //步骤三：数据的处理
        SingleOutputStreamOperator<WordAndOne> results = dataStream
                .flatMap(new MyFlatMap()).setParallelism(2)
                .keyBy(tuple -> tuple.word)
                .sum("one").setParallelism(2);
        //步骤四：数据的输出
        results.print().setParallelism(2);
        //步骤五：启动程序
        env.execute("wordcount");
    }

    //分割字符串，生成(word,1)
    public static class MyFlatMap implements FlatMapFunction<String,WordAndOne>{
        @Override
        public void flatMap(String s, Collector<WordAndOne> collector) throws Exception {
            String[] words = s.split(",");
            for (String word : words) {
                collector.collect(new WordAndOne(word, 1));

            }
        }
    }

    public static class WordAndOne{
        private String word;
        private Integer one;

        public WordAndOne() {
        }

        public WordAndOne(String word, Integer one) {
            this.word = word;
            this.one = one;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public Integer getOne() {
            return one;
        }

        public void setOne(Integer one) {
            this.one = one;
        }

        @Override
        public String toString() {
            return "WordAndOne{" +
                    "word='" + word + '\'' +
                    ", one=" + one +
                    '}';
        }
    }
}
